package com.atguigu.ct.consumer.coprocessor;

import com.atguigu.ct.common.bean.BaseDao;
import com.atguigu.ct.common.constant.Names;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.RegionObserver;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.wal.WALEdit;

import java.io.IOException;
import java.util.Optional;

/**
 * Description: 使用协处理器保存被叫用户的数据
 *              协处理器的使用
 *              1.创建类
 *              2.让表知道协处理类（和表有关联）
 *              3.将项目打成jar包放到Hbase中（关联的jar包也需要发布），并且需要分发
 * @Author: yyx
 * @Create: 2023/5/4 15:47
 * @Version: 8
 */
public class InsertCalleeCoprocessor implements RegionObserver, RegionCoprocessor {
    // prePut
    // doPut:模板方法设计模式
        //存在父子类
        //父类搭建算法的骨架
        // 1.tel取用户代码 2.时间取年月 3.异或运算 4.hash散列
        //子类重写算法的细节
        // do1.tel取后4位 do2.202305 do3.^ do4.%/&
    // postPut

    /**
     * 保存主叫用户数据之后，由Hbase自动保存被叫用户数据
     * @param c
     * @param put
     * @param edit
     * @throws IOException
     */
    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit) throws IOException {
        //获取表
        Table table = c.getEnvironment().getConnection().getTable(TableName.valueOf(Names.TABLE.getValue()));
        String rowkey = Bytes.toString(put.getRow());//主叫用户的rowkey
        String[] values = rowkey.split("_");
        CoprocessorDao coprocessorDao = new CoprocessorDao();
        String call1 = values[1];
        String call2 = values[3];
        String calltime = values[2];
        String duration = values[4];
        String flg = values[5];

        //保存数据(!!!只有主叫用户保存后才需要触发被叫用户的保存)
        if("1".equals(flg)){
            String calleeRowkey = coprocessorDao.getRegionNm(call2,calltime) + "_" + call2 + "_" + calltime + "_" + call1 + "_" + duration + "_0";
            Put callePut = new Put(Bytes.toBytes(calleeRowkey));
            byte[] calleeFamily = Bytes.toBytes(Names.CF_CALLEE.getValue());
            callePut.addColumn(calleeFamily,Bytes.toBytes("call1"),Bytes.toBytes(call2));
            callePut.addColumn(calleeFamily,Bytes.toBytes("call2"),Bytes.toBytes(call1));
            callePut.addColumn(calleeFamily,Bytes.toBytes("calltime"),Bytes.toBytes(calltime));
            callePut.addColumn(calleeFamily,Bytes.toBytes("duration"),Bytes.toBytes(duration));
            callePut.addColumn(calleeFamily,Bytes.toBytes("flg"),Bytes.toBytes("0"));
            table.put(callePut);
            //关闭表
            table.close();
        }
    }

    @Override
    public Optional<RegionObserver> getRegionObserver() {
        return Optional.of(this);
    }

    private class CoprocessorDao extends BaseDao {
        public int getRegionNm(String tel,String time){
            return genRegionNum(tel,time);
        }
    }
}
